package com.nehc.nettyserver.service.impl;

import com.nehc.nettyserver.common.model.Message;
import com.nehc.nettyserver.service.MessageService;
import com.nehc.nettyserver.service.cache.CacheService;
import com.nehc.nettyserver.service.storage.StorageService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 消息服务实现类
 * 
 * @author NEHC
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class MessageServiceImpl implements MessageService {
    
    private final StorageService storageService;
    private final CacheService cacheService;
    
    /**
     * 内存中的最近消息缓存，按用户ID分组
     */
    private final Map<String, List<Message>> userMessageCache = new ConcurrentHashMap<>();
    
    /**
     * 消息统计计数器
     */
    private final Map<Message.Type, Long> messageTypeCounter = new ConcurrentHashMap<>();
    private final Map<Integer, Long> messageHourCounter = new ConcurrentHashMap<>();
    private long todayMessageCount = 0;
    
    @Override
    public boolean saveMessage(Message message) {
        if (message == null || message.getId() == null) {
            return false;
        }
        
        // 缓存消息
        cacheService.put(CacheService.CacheType.MESSAGE, message.getId(), message);
        
        // 更新用户消息缓存
        if (message.getType() == Message.Type.CHAT) {
            // 发送者的消息
            String fromUserId = message.getFrom();
            if (fromUserId != null) {
                updateUserMessageCache(fromUserId, message);
            }
            
            // 接收者的消息
            if (message.getPayload().containsKey("to")) {
                String toUserId = message.getPayload().get("to").toString();
                if (toUserId != null) {
                    updateUserMessageCache(toUserId, message);
                }
            }
        }
        
        // 更新统计计数器
        updateMessageStats(message);
        
        // 异步持久化消息
        asyncSaveMessage(message);
        
        return true;
    }
    
    /**
     * 异步保存消息到持久化存储
     */
    @Async
    public CompletableFuture<Boolean> asyncSaveMessage(Message message) {
        try {
            return storageService.saveAsync(message.getId(), message);
        } catch (Exception e) {
            log.error("保存消息失败: {}", e.getMessage(), e);
            return CompletableFuture.completedFuture(false);
        }
    }
    
    /**
     * 更新用户消息缓存
     */
    private void updateUserMessageCache(String userId, Message message) {
        userMessageCache.computeIfAbsent(userId, k -> Collections.synchronizedList(new ArrayList<>()))
                .add(message);
        
        // 限制每个用户的缓存消息数量
        List<Message> messages = userMessageCache.get(userId);
        if (messages.size() > 100) {
            synchronized (messages) {
                if (messages.size() > 100) {
                    userMessageCache.put(userId, new ArrayList<>(messages.subList(messages.size() - 100, messages.size())));
                }
            }
        }
    }
    
    /**
     * 更新消息统计
     */
    private void updateMessageStats(Message message) {
        // 按类型统计
        messageTypeCounter.compute(message.getType(), (k, v) -> (v == null) ? 1 : v + 1);
        
        // 按小时统计
        Calendar calendar = Calendar.getInstance();
        calendar.setTimeInMillis(message.getTimestamp());
        int hour = calendar.get(Calendar.HOUR_OF_DAY);
        messageHourCounter.compute(hour, (k, v) -> (v == null) ? 1 : v + 1);
        
        // 总数统计
        todayMessageCount++;
    }

    @Override
    public Message getMessage(String messageId) {
        // 先从缓存获取
        Message message = cacheService.get(CacheService.CacheType.MESSAGE, messageId);
        if (message != null) {
            return message;
        }
        
        // 从存储获取
        try {
            Optional<Message> optionalMessage = storageService.get(messageId, Message.class);
            if (optionalMessage.isPresent()) {
                message = optionalMessage.get();
                // 放入缓存
                cacheService.put(CacheService.CacheType.MESSAGE, messageId, message);
                return message;
            }
        } catch (IOException e) {
            log.error("获取消息失败: {}", e.getMessage(), e);
        }
        
        return null;
    }

    @Override
    public List<Message> getUserMessages(String userId, int limit) {
        List<Message> cachedMessages = userMessageCache.getOrDefault(userId, Collections.emptyList());
        
        if (cachedMessages.size() >= limit) {
            // 如果缓存里有足够多的消息，直接返回
            int fromIndex = Math.max(0, cachedMessages.size() - limit);
            return new ArrayList<>(cachedMessages.subList(fromIndex, cachedMessages.size()));
        }
        
        // 否则需要从存储获取更多消息
        // TODO: 实现从存储获取历史消息
        
        return cachedMessages;
    }

    @Override
    public MessageStats getMessageStats(long startTime, long endTime) {
        MessageStats stats = new MessageStats();
        
        // 设置总消息数
        stats.setTotalMessages(todayMessageCount);
        
        // 设置按类型统计的消息数
        stats.setMessagesByType(new HashMap<>(messageTypeCounter));
        
        // 设置按小时统计的消息数
        stats.setMessagesByHour(new HashMap<>(messageHourCounter));
        
        // 设置活跃用户数和连接数
        stats.setActiveUsers(userMessageCache.size());
        stats.setActiveConnections(cacheService.getAll(CacheService.CacheType.CONNECTION).size());
        
        return stats;
    }
    
    /**
     * 获取消息格式化时间
     */
    private String formatMessageTime(long timestamp) {
        Instant instant = Instant.ofEpochMilli(timestamp);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
                .withZone(ZoneId.systemDefault());
        return formatter.format(instant);
    }

    /**
     * 获取客户端消息历史
     */
    @Override
    public List<Message> getClientMessageHistory(String clientId, int offset, int limit) {
        // 从缓存中获取该客户端的消息
        List<Message> cachedMessages = getClientCachedMessages(clientId);
        
        // 如果缓存数量足够，直接返回子列表
        if (cachedMessages.size() >= offset + limit) {
            int endIndex = Math.min(offset + limit, cachedMessages.size());
            return cachedMessages.subList(offset, endIndex);
        }
        
        // 缓存不足，需要从存储服务获取补充
        try {
            // 从存储服务获取消息历史
            List<Message> storedMessages = storageService.getClientMessages(clientId, offset, limit);
            
            // 合并去重
            Set<String> cachedIds = cachedMessages.stream()
                    .map(Message::getId)
                    .collect(Collectors.toSet());
            
            List<Message> result = new ArrayList<>(cachedMessages);
            for (Message message : storedMessages) {
                if (!cachedIds.contains(message.getId())) {
                    result.add(message);
                }
            }
            
            // 按时间排序
            result.sort(Comparator.comparing(Message::getTimestamp).reversed());
            
            // 返回指定范围
            int endIndex = Math.min(offset + limit, result.size());
            return offset < result.size() ? result.subList(offset, endIndex) : new ArrayList<>();
        } catch (Exception e) {
            log.error("Failed to get client message history for clientId: {}", clientId, e);
            // 返回缓存内容
            int endIndex = Math.min(offset + limit, cachedMessages.size());
            return offset < cachedMessages.size() ? cachedMessages.subList(offset, endIndex) : new ArrayList<>();
        }
    }

    /**
     * 获取客户端缓存的消息
     */
    private List<Message> getClientCachedMessages(String clientId) {
        // 尝试从用户消息缓存获取
        List<Message> messages = new ArrayList<>();
        userMessageCache.values().forEach(userMessages -> {
            userMessages.forEach(message -> {
                // 如果消息涉及到该客户端，则添加到结果集
                if (clientId.equals(message.getSenderId()) || clientId.equals(message.getReceiverId())) {
                    messages.add(message);
                }
            });
        });
        
        // 按时间戳降序排序
        messages.sort(Comparator.comparing(Message::getTimestamp).reversed());
        return messages;
    }

    // 修改异步保存方法的调用
    private void asyncSaveMessage(Message message) {
        try {
            // 使用存储服务保存消息
            storageService.saveMessage(message);
        } catch (Exception e) {
            log.error("Failed to save message asynchronously: {}", message.getId(), e);
        }
    }
} 